#pragma once

#include <brpc/closure_guard.h>
#include <brpc/controller.h>
#include <brpc/server.h>
#include <cstddef>
#include <cstdlib>
#include <etcd/Client.hpp>
#include <brpc/channel.h>
#include <librdkafka/rdkafkacpp.h>
#include <log.h>
#include <memory>
#include <odb/mysql/database.hxx>
#include <optional>
#include <string>
#include <vector>

#include "Util.hpp"
#include "brpc.h"
// #include "storage.pb.h"
#include "kafka_consummer.hpp"
#include "user.pb.h"
#include "UserOperator.hpp"
#include "etcd.hpp"
#include "redis.h"
#include "mysql.h"
#include "etcd_path.h"
#include "UserOperator.hpp"
#include "user_action.pb.h"
#include "verify_code.hpp"
#include "user_redis.h"
#include "es.h"

namespace lyt{
    class UserServiceImpl : public user::UserService{
    private:
        // user data mysql
        UserData::ptr _user_mysql;
        // redis
        UserRedis::ptr _user_redis;
        // es
        esclient::ptr _esclient;
        index::ptr _index;
        insert::ptr _insert;
    private:
        // 存储的数据 返回消息 存储成功返回的uuid
        /* 
        bool Save(const std::string& data,std::string& message,std::string& uuid){
            file::FileStorageService_Stub stub(_channel.get());
            file::SaveFileRequest savefilerequest;
            file::SaveFileResponse savefileresponse;
            savefilerequest.set_file_data(data);
            brpc::Controller cntl;
            stub.SaveFile(&cntl, &savefilerequest, &savefileresponse, nullptr);
            if (cntl.Failed()) {
                message = "存储服务不可用";
                return false;
            }
            if (savefileresponse.success() == false) {
                message = savefileresponse.message();
                return false;
            }
            uuid = savefileresponse.file_uuid();
            return true;
        }
        // 文件的名称（uuid） 返回的消息 文件的内容
         
        bool GetUsrAvatar(const std::string uuid,std::string& message,std::string& data){
            file::FileStorageService_Stub stub(_channel.get());
            file::GetFileRequest getfilerequest;
            file::GetFileResponse getfileresponse;
            getfilerequest.set_file_uuid(uuid);
            brpc::Controller cntl;
            stub.ReadFile(&cntl, &getfilerequest, &getfileresponse, nullptr);
            if (cntl.Failed()) {
                message = "存储服务不可用";
                debuglog << "存储服务不可用";
                return false;
            }
            if(!getfileresponse.success()){
                message = getfileresponse.message();
                return false;
            }
            data = getfileresponse.file_data();
            debuglog << "get user avatar success";
            return true;
            
        }
        bool DelUsrAvatar(const std::string uuid,std::string& message){
            file::FileStorageService_Stub stub(_channel.get());
            file::DelFileRequest delfilerequest;
            file::DelFileResponse delfileresponse;
            delfilerequest.set_file_uuid(uuid);
            brpc::Controller cntl;
            stub.DelFile(&cntl, &delfilerequest, &delfileresponse, nullptr);
            if (cntl.Failed()) {
                message = "存储服务不可用";
                return false;
            }
            return true;   
        }*/
    public:
        using ptr = std::shared_ptr<UserServiceImpl>;

        UserServiceImpl(const shared_ptr<odb::mysql::database>& odb,
                        const std::shared_ptr<sw::redis::Redis>& redis,
                        const std::shared_ptr<esclient> esclient) 
            :_user_mysql(std::make_shared<UserData>(odb)),
            _user_redis(std::make_shared<UserRedis>(redis)),
            _esclient(esclient),
            _index(std::make_shared<index>(_esclient)),
            _insert(std::make_shared<insert>(_esclient)){

            _index->add_keyword_field("email")
                    .add_keyword_field("avatar")
                    .add_text_field("name")
                    .add_integer_field("solved_count")
                    .add_integer_field("total_problems")
                    .add_integer_field("solved_problems")
                    .run("user");
            
        }
        void SendCode(::google::protobuf::RpcController* controller,
                        const ::user::VerificationCodeRequest* request,
                        ::user::VerificationCodeResponse* response,
                        ::google::protobuf::Closure* done){
            brpc::ClosureGuard done_guard(done);

            if(request->email().empty()){
                response->set_success(false);
                response->set_message("邮箱不能为空~");
                return;
            }
            // 匹配邮箱格式是否正确 
            const std::regex emailRegex(R"(^[^\s@]+@[^\s@]+\.[^\s@]+$)");
            if(!std::regex_match(request->email(), emailRegex)){
                response->set_success(false);
                response->set_message("邮箱格式错误~");
                return;
            }

            std::string code; // 验证码
            std::string message;
            EmailSender ems;
            if(!ems.sendEmail(request->email(), code)){
                response->set_success(false);
                response->set_message("邮箱不存在");
                return;
            }
            // 生成uuid将验证码id存入redis中
            std::string code_id = lyt::Generate_UUID::generate_uuid();
            debuglog << "code_id:" << code_id;
            if(!_user_redis->SetRedisCode(code_id, code,300)){
                errorlog << "set redis error!";
                abort();
            }
            response->set_success(true);
            response->set_message("验证码发送成功~");
            response->set_code_id(code_id);
        }
        void Register(::google::protobuf::RpcController* controller,
                        const ::user::RegisterRequest* request,
                        ::user::RegisterResponse* response,
                        ::google::protobuf::Closure* done){
            // 1. 判断验证码是否正确
            // 2. 根据邮箱查询注册的邮箱是否被使用
            // 3. 根据用户名查询用户名是否被占用
            // 4. 如果用户上传了头像，调用文件存储服务保存
            // 5. 存储用户信息到数据库
            //     1. 用户头像上传成功 -> 存储用户信息
            // 6. 构造响应并返回
            brpc::ClosureGuard done_guard(done);
            auto result = _user_redis->GetCode(request->code_id());
            debuglog << "request->code_id()" << request->code_id();
            if(result == std::nullopt){
                response->set_success(false);
                response->set_message("验证码已经过时！");
                return;
            }
            if(*result != request->code()){
                response->set_success(false);
                response->set_message("验证码输入错误！");
                return;
            }
            std::vector<std::vector<std::string>> info;
            if(_user_mysql->query_user_email(request->email())){
                // email被占用
                response->set_success(false);
                response->set_message("该邮箱已经被注册，请更换邮箱后再试");
                return;
            }

            if(_user_mysql->exist_username(request->username())){
                // user_name被占用
                response->set_success(false);
                response->set_message("该用户名已经被注册，请更换用户名后再试");
                return;
            }

            if (request->avatar().empty()) { 
                response->set_success(false);
                response->set_message("头像不能为空");
                return;
            }
            std::string version = Generate_UUID::generate_uuid(); // 用户注册时维护一个version
            if(!_user_mysql->insert_user(request->email(),request->avatar(),request->username(),request->password(),version)){
                response->set_success(false);
                response->set_message("服务器繁忙");
                return;
            }
            // std::string message,data;
            // GetUsrAvatar(avatar_uuid, message, data);
            // if(data.empty()) debuglog << "data empty!";
            // if(request->avatar() == data) debuglog << "存储读取没有改变数据";
            // 插入用户之后，也要把UserStatsSummary的数据插入mysql与redis
            UserStatsSummary uss(request->email(),0,0,0,0,0,0,0);
            // 插入数据库
            if(!_user_mysql->insert_userstatssummary(uss)){
                response->set_success(false);
                response->set_message("服务器繁忙");
                return;
            }
            // 插入redis
            if(!_user_redis->SetRedisStatsSummary(request->email(), 0,0,0,0,0,0,0,0)){
                response->set_success(false);
                response->set_message("服务器繁忙");
                return;
            }
            // 把用户插入es
            _insert->set_value("email",request->email())
                    .set_value("avatar",request->avatar())
                    .set_value("name",request->username())
                    .set_value("solved_count",0)
                    .set_value("total_problems",0)
                    .set_value("solved_problems",0);
                    
            response->set_success(true);
            response->set_message("注册成功，开启刷题之旅吧~");
        }
        void Login(::google::protobuf::RpcController* controller,
                        const ::user::LoginRequest* request,
                        ::user::LoginResponse* response,
                        ::google::protobuf::Closure* done){
            brpc::ClosureGuard done_gurad(done);
            if(!_user_mysql->query_user_email(request->email())){
                response->set_success(false);
                response->set_message("用户邮箱输入错误！");
                return;
            }else{
                // 查询到邮箱，验证密码是否正确
                std::string pwd;
                _user_mysql->query_user_password(request->email(), pwd);
                if(pwd != request->password()){
                    // 密码错误
                    response->set_success(false);
                    response->set_message("密码错误，请输入正确密码!");
                    return;
                }
                // 构建token
                std::string token = Generate_Token::generate_token();
                // 如果有token信息，那就fresh一下，没有才创建

                // 存入redis
                if(!_user_redis->SetToken(token, request->email(), 60*60*24*7)){//过期时间7天
                    controller->SetFailed("failed to set token!");
                    return;
                }
                user::UserStats userstats;
                User user;
                _user_mysql->query_user(request->email(), user);
                userstats.set_email(user.email);
                userstats.set_avatar(user.avatar);
                userstats.set_name(user.username);
                userstats.set_solved_count(_user_redis->GetSovledCount(user.email));
                userstats.set_total_problems(_user_redis->GetTotalProblems(user.email));
                std::vector<int> solved_idxs =_user_redis->GetResolvedProblems(user.email);
                for(auto e : solved_idxs){
                    userstats.add_solved_problems(e);
                }
                response->set_token(token);
                // 这样做是将UserStats对象userstats所有权转移到response的stats字段，
                // 但是，这个对象除了作用域就销毁，就会导致悬空指针，非常危险
                // response->set_allocated_stats(&userstats);
                // 将userstats的值拷贝一份，安全做法
                *response->mutable_stats() = userstats;
                response->set_version(user.version);
                if(!_user_redis->AddUserVersion(request->email(), user.version)){
                    response->set_success(false);
                    response->set_message("服务器错误!");
                }
                response->set_success(true);
                response->set_message("登录成功！");
                response->set_token(token);
            }
        }
        void CodeLogin(::google::protobuf::RpcController* controller,
                        const ::user::CodeLoginRequest* request,
                        ::user::LoginResponse* response,
                        ::google::protobuf::Closure* done){
            brpc::ClosureGuard done_gurad(done);
            debuglog << "In CodeLogin";
            debuglog << request->email();
            if(!_user_mysql->query_user_email(request->email())){
                debuglog << "request email empty";
                response->set_success(false);
                response->set_message("用户邮箱输入错误");
                return;
            }else{
                // 查询到邮箱，验证密码是否正确
                // info = [email,avatar,username,password,version]
                // 根据验证码id来对比验证码
                debuglog << "request code_id:" << request->code_id();
                auto result = _user_redis->GetCode(request->code_id());
                
                if(result == std::nullopt){
                    debuglog << "验证码已经过时";
                    response->set_success(false);
                    response->set_message("验证码已经过时！");
                    return;
                }
                if(*result != request->code()){
                    debuglog << "验证码输入错误";

                    response->set_success(false);
                    response->set_message("验证码输入错误！");
                    return;
                }
                // 构建token
                std::string token = Generate_Token::generate_token();
                // 存入redis
                if(!_user_redis->SetToken(token, request->email(), 60*60*24*7)){//过期时间7天
                    debuglog << "failed to set toekn 7 days";
                    controller->SetFailed("failed to set token!");
                    abort();
                }
                debuglog << "set token success,begin to set userstats";
                user::UserStats userstats;
                User user;
                _user_mysql->query_user(request->email(), user);
                userstats.set_email(user.email);
                userstats.set_avatar(user.avatar);
                userstats.set_name(user.username);
                userstats.set_solved_count(_user_redis->GetSovledCount(user.email));
                userstats.set_total_problems(_user_redis->GetTotalProblems(user.email));
                std::vector<int> solved_idxs =_user_redis->GetResolvedProblems(user.email);
                for(auto e : solved_idxs){
                    userstats.add_solved_problems(e);
                }
                debuglog << "set userstatus success";
                *response->mutable_stats() = userstats;
                debuglog << "success return code login";
                response->set_version(user.version);
                if(!_user_redis->AddUserVersion(request->email(), user.version)){
                    response->set_success(false);
                    response->set_message("服务器错误!");
                }
                response->set_success(true);
                response->set_message("登录成功！");
                response->set_token(token);
            }
        }
        void GetUserstats(::google::protobuf::RpcController* controller,
                        const ::user::UserStatsRequest* request,
                        ::user::UserStatsResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 验证token合法性
                            2. 填写stats，根据请求的是否是自己决定是否填写solved_problem
                            3. 请求自己的信息需要返回version，别人的不需要返回version
                            4. 
                            */
            brpc::ClosureGuard done_gurad(done);
            auto opt = _user_redis->GetUserId(request->token());
            if(opt == std::nullopt){
                debuglog << "redis没有对应用户信息";
                controller->SetFailed("Invalid token!");
                return;
            }
            std::string request_email = *opt;
            std::string target_email = request->email();
            if (target_email == "" || target_email == request_email) {
                // 获取自己的用户信息

                // 从redis中获取到version

                std::string version = _user_redis->GetUserVersion(request_email);
                if (request->version() == "" || request->version() != version) {
                    // 如果用户请求的version为空，或者和redis中的version不一致，需要填充新的用户信息和version版本
                    // 查询用户基本信息（email, avatar uuid, name）
                    User user;
                    if (!_user_mysql->query_user(request_email, user)) {
                        debuglog << "User not found!";
                        controller->SetFailed("User not found!");
                        return;
                    }
                    user::UserStats userstats;
                    userstats.set_email(user.email); 
                    userstats.set_avatar(user.avatar);
                    userstats.set_name(user.username);
                    userstats.set_solved_count(_user_redis->GetSovledCount(request_email));
                    userstats.set_total_problems(_user_redis->GetTotalProblems(request_email));
                    vector<int> solved_problems = _user_redis->GetResolvedProblems(request_email);
                    for(auto e : solved_problems){
                        userstats.add_solved_problems(e);
                    }
                    *response->mutable_stats() = userstats;
                    response->set_version(user.version);
                }else {
                    // 填充version信息，用于告诉前端信息是最新的，不需要更新
                    response->set_version(_user_redis->GetUserVersion(request_email));
                }     
            }else {
                // 填充其他用户的基本信息
                User user;
                if(target_email.empty()){
                    response->set_success(false);
                    response->set_message("请求的用户不能为空!");
                    return;
                }
                if (!_user_mysql->query_user(target_email, user)) {
                    debuglog << "User not found!";
                    controller->SetFailed("User not found!");
                    return;
                }
                user::UserStats userstats;
                userstats.set_email(user.email); 
                userstats.set_avatar(user.avatar);
                userstats.set_name(user.username);
                userstats.set_solved_count(_user_redis->GetSovledCount(target_email));
                userstats.set_total_problems(_user_redis->GetTotalProblems(target_email));
                *response->mutable_stats() = userstats;
                response->set_version(user.version);
            }
            response->set_success(true);
            response->set_message("获取用户信息成功");
        }
        void GetUserstatsSummary(::google::protobuf::RpcController* controller,
                        const ::user::UserStatsSummaryRequest* request,
                        ::user::UserStatsSummaryResponse* response,
                        ::google::protobuf::Closure* done){
            brpc::ClosureGuard done_guard(done);

            auto opt = _user_redis->GetUserId(request->token());

            if(opt == std::nullopt){
                debuglog << "token 查询不到对应的用户信息";
                response->set_success(false);
                response->set_message("用户未登录!");
                return;
            }
            std::string email = *opt;
            user::UserStatsSummary uss;
            std::vector<std::string> summary;
            if(!_user_redis->GetRedisStatsSummary(request->email(), summary)){
                controller->SetFailed("Get user stats summary failed!");
                return;
            }
            if(summary.empty()){
                response->set_success(false);
                response->set_message("没有对应用户!");
                return;
            }
            uss.set_solved_problems(stoi(summary[0]));
            uss.set_attempting_problems(stoi(summary[1]));
            uss.set_published_blogs(stoi(summary[2]));
            uss.set_draft_blogs(stoi(summary[3]));
            uss.set_following(stoi(summary[4]));
            uss.set_followers(stoi(summary[5]));
            uss.set_mutual_friends(stoi(summary[6]));
            uss.set_blog_likes_number(stoi(summary[7]));
            *response->mutable_stats() = uss;
            response->set_message("获取成功");
            response->set_success(true);
        }
        void Logout(::google::protobuf::RpcController* controller,
                        const ::user::LogoutRequest* request,
                        ::user::LogoutResponse* response,
                        ::google::protobuf::Closure* done){
            /*
            1. 查询token，删除token
            */
            brpc::ClosureGuard done_guard(done);
            if(!_user_redis->exists(request->token())){
                response->set_success(false);
                response->set_message("用户已登出，请勿重复操作!");
            }
            auto email = _user_redis->GetUserId(request->token());
            if(!_user_redis->DeleteUserVersion(*email)){
                controller->SetFailed("DeleteUserVersion failed");
                return;
            }
            if(!_user_redis->DelToken(request->token())){
                controller->SetFailed("redis delete token failed");
                return;
            }
            response->set_success(true);
            response->set_message("退出登录成功");
        }
        void Updateusername(::google::protobuf::RpcController* controller,
                        const ::user::UpdateUsernameRequest* request,
                        ::user::UpdateUsernameResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 查询token，查看token是否合法
                            2. 根据token查询出email
                            3. 用户名不能和原来一致或者和其他人重复  
                            4. 修改用户名，更新redis以及数据库信息，重新生成版本号                     
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());   
            if(!opt){
                debuglog << "Not have user in redis";
                controller->SetFailed("Invalid token");
                return;
            }
            std::string email = *opt;
            User user;
            _user_mysql->query_user(email, user);
            // info: [email,avatar,username,password,version]
            if(user.username == request->new_username()){
                // 新用户名和原来的用户名一致
                response->set_success(false);
                response->set_message("新用户名和旧用户名不能相同!");
                return;
            }
            std::string name;
            if(!_user_mysql->query_user_name(email,name)){
                controller->SetFailed("Failed to query username");
                return;
            }
            if(name == request->new_username()){
                response->set_success(false);
                response->set_message("新用户名与其它用户重复!");
                return;
            }
            if(!_user_mysql->update_username(email, request->new_username()))
            {
                controller->SetFailed("failed to update username");
                return;
            }
            // 生成新的version
            std::string version = Generate_UUID::generate_uuid();
            if(!_user_mysql->update_version(email, version)){
                controller->SetFailed("failed to update user version");
                return;
            }
            response->set_success(true);
            response->set_message("用户名修改成功");
        }
        void Updateuseravatar(::google::protobuf::RpcController* controller,
                        const ::user::UpdateAvatarRequest* request,
                        ::user::UpdateAvatarResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 查询token，查看token是否合法
                            2. 根据token查询用户名
                            3. 调用文件存储服务，更新数据库数据
                            4. 更新用户版本
                            5. 删除原来的旧头像文件                   
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());   
            if(!opt){
                debuglog << "Invalid token";
                controller->SetFailed("Invalid token");
                response->set_success(false);
                response->set_message("修改头像失败，用户未登录");
                return;
            }
            std::string email = *opt;
            User user;
            _user_mysql->query_user(email, user);
            // info: [email,avatar,username,password,version]
            // 将之前的头像文件名保存一份
            std::string before_avatar_uuid = user.avatar;

            if(request->avatar().empty()){
                response->set_success(false);
                response->set_message("头像不能为空");
                return;
            }
            if(!_user_mysql->update_avartar(email, request->avatar())){
                debuglog << "update_avartar failed";
                controller->SetFailed("failed to update avartar");
                return;
            }
            std::string uuid = Generate_UUID::generate_uuid();
            if(!_user_mysql->update_version(email, uuid)){
                debuglog << "update_version failed";
                controller->SetFailed("failed to update version");
                return;
            }
            response->set_success(true);
            response->set_message("新头像设置成功");
            response->set_avatar(request->avatar()); 
            debuglog << "Out Updateuseravatar";
        }
        void UpdatePassword(::google::protobuf::RpcController* controller,
                        const ::user::ChangePasswordRequest* request,
                        ::user::ChangePasswordResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 确认token有效性
                            2. 判断当前密码是否与sql中密码相同
                            2. 新密码必须是大小写字母，数字，_,*,/,.,^,&,$,!
                            3. 新密码不能和旧密码相同
                            4. 更新数据库密码（不需要重新获取用户版本）
                            5. 构造响应数据     
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());   
            if(!opt){
                controller->SetFailed("Invalid token");
                response->set_success(false);
                response->set_message("修改密码失败，用户未登录");
                return;
            }
            std::string email =  *opt;
            std::string before_pwd;
            _user_mysql->query_user_password(email,before_pwd);
            if(before_pwd != request->current_password()){
                response->set_success(false);
                response->set_message("旧密码错误，请重新输入");
                return;
            }

            std::regex pattern(R"(^[A-Za-z0-9_\*/\.\^&\$\!]+$)");
            if(!std::regex_match(request->new_password(), pattern)){
                response->set_success(false);
                response->set_message("密码必须是大小写字母、数字或\"_*/.^&$!\"中的字符");
                return;
            }
            
            if(before_pwd == request->new_password()){
                response->set_success(false);
                response->set_message("新密码不能与旧密码相同!");
                return;
            }

            if(!_user_mysql->update_password(email, request->new_password())){
                controller->SetFailed("failed to update password!");
                return;
            }
            response->set_success(true);
            response->set_message("密码修改成功~");
        }
        void UpdatePasswordByEmail(::google::protobuf::RpcController* controller,
                        const ::user::ChangePasswordByEmailRequest* request,
                        ::user::ChangePasswordResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 验证token
                            2. 使用codeid从redis中取验证码和请求的验证码进行对比，对比失败则返回
                            3. 新密码必须是大小写字母，数字，_,*,/,.,^,&,$,!
                            4. 新密码不能和旧密码相同
                            5. 更新数据库密码（不需要重新获取用户版本）
                            6. 构造响应数据                    
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());
            if(!opt){
                response->set_success(false);
                response->set_message("修改密码失败，用户未登录");
                return;
            }
            std::string email = *opt;
            auto result = _user_redis->GetCode(request->code_id());
            
            if(result == std::nullopt){
                response->set_success(false);
                response->set_message("验证码已经过时！");
                return;
            }
            if(*result != request->code()){
                response->set_success(false);
                response->set_message("验证码输入错误！");
                return;
            }

            std::regex pattern(R"(^[A-Za-z0-9_\*/\.\^&\$\!]+$)");
            if(!std::regex_match(request->new_password(), pattern)){
                response->set_success(false);
                response->set_message("密码必须是大小写字母、数字或\"_*/.^&$!\"中的字符");
                return;
            }
            std::string before_pwd;
            _user_mysql->query_user_password(email,before_pwd);
            if(before_pwd == request->new_password()){
                response->set_success(false);
                response->set_message("新密码不能与旧密码相同!");
                return;
            }

            if(!_user_mysql->update_password(email, request->new_password())){
                controller->SetFailed("failed to update password!");
                return;
            }
            response->set_success(true);
            response->set_message("密码修改成功~");
        }
        void GetFollowsStatus(::google::protobuf::RpcController* controller,
                        const ::user::GetFollowStatusRequest* request,
                        ::user::GetFollowStatusResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 验证token
                            2. email为空，差错处理
                            3. 取出token的邮箱，去redis中查看与request中的email关注情况
                            4. 如果redis中查不到，就去mysql中去查
                            5. 构造响应
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());
            if(!opt){
                response->set_success(false);
                response->set_message("请求邮箱错误");
                return;
            }
            if(!_user_mysql->query_user_email(request->email())){
                response->set_success(false);
                response->set_message("请求用户不存在");
                return;
            }
            std::string email = *opt;
            // 自己是否关注了对方
            bool isfollow = _user_redis->IsFollow(email, request->email());
            if(!isfollow){
                // redis中查不到，去sql中
                isfollow = _user_mysql->query_user_follow(email, request->email());
            }
            response->set_following(isfollow);
            // 对方是否关注了自己
            isfollow = _user_redis->IsFollow(request->email(), email);
            if(!isfollow){
                // redis中查不到，去sql中
                isfollow = _user_mysql->query_user_follow(request->email(), email);
            }

            response->set_followers(isfollow);
            response->set_success(true);
            response->set_message("获取成功");
        }
        void Follow(::google::protobuf::RpcController* controller,
                        const ::user::FollowRequest* request,
                        ::user::FollowResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 验证token
                            2. 验证要关注的用户是否存在
                            3. 关注用户,同时设置到redis与mysql中
                            4. 构造响应
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());
            if(!opt){
                response->set_success(false);
                response->set_message("");
                return;
            }
            std::string email = *opt;
            if(!_user_mysql->query_user_email(request->email())){
                response->set_success(false);
                response->set_message("关注的用户升华了!");
                return;
            }
            if(!_user_redis->AddFollow(email, request->email()) || !_user_mysql->insert_user_follow(email, request->email())){
                response->set_success(false);
                response->set_message("该用户不想让你关注~");
                return;
            }
            response->set_success(true);
            response->set_message("关注成功,开启窥视~");
        }
        void UnFollow(::google::protobuf::RpcController* controller,
                        const ::user::UnfollowRequest* request,
                        ::user::UnfollowResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 验证token
                            2. email为空，差错处理
                            3. 取出token的邮箱，去redis中查看与request中的email关注情况
                            4. 构造响应
                            */
            brpc::ClosureGuard done_guard(done);
            auto opt = _user_redis->GetUserId(request->token());
            if(!opt){
                response->set_success(false);
                response->set_message("");
                return;
            }
            std::string email = *opt;
            if(!_user_mysql->query_user_email(request->email())){
                response->set_success(false);
                response->set_message("取消关注的用户升华了!");
                return;
            }
            if(!_user_redis->UnFollow(email, request->email()) || !_user_mysql->delete_user_follow(email, request->email())){
                response->set_success(false);
                response->set_message("该用户不想让你取消关注~");
                return;
            }
            response->set_success(true);
            response->set_message("取消关注成功,结束窥视~");
        }
        void GetUserInfo(::google::protobuf::RpcController* controller,
                        const ::user::GetUserInfoRequest* request,
                        ::user::GetUserInfoResponse* response,
                        ::google::protobuf::Closure* done){
                            /*
                            1. 验证token
                            2. email为空，差错处理
                            3. 查看请求的用户是否存在
                            4. 填写信息
                            5. 构造响应
                            */
            brpc::ClosureGuard done_guard(done);
            auto handler = [&](bool success, const std::string& errmsg = ""){
                response->set_success(success);
                response->set_message(errmsg);
                return ;
            };
            auto opt = _user_redis->GetUserId(request->token());
            if(!opt){
                response->set_success(false);
                response->set_message("");
                return;
            }
            if(!_user_mysql->query_user_email(request->email())){
                response->set_success(false);
                response->set_message("请求的用户信息不存在");
                return;
            }
            std::string avartar_path,name;
            _user_mysql->query_user_avatar(request->email(), avartar_path);
            _user_mysql->query_user_name(request->email(), name);
            response->set_success(true);
            response->set_avatar(avartar_path);
            response->set_email(request->email());
            response->set_name(name);
            response->set_message("获取成功");
        }
        void GetDefaultAvatar(::google::protobuf::RpcController* controller,
            const ::user::GetDefaultAvatarRequest* request,
            ::user::GetDefaultAvatarResponse* response,
            ::google::protobuf::Closure* done){
                /*
                1. 
                */
            brpc::ClosureGuard done_guard(done);
            auto handler = [&](bool success, const std::string& errmsg = ""){
                response->set_success(success);
                response->set_message(errmsg);
                return ;
            };
            std::vector<std::string> res;
            _user_mysql->query_default_avatar(res);
            for(auto e : res){
                response->add_avatars(e);
            }
            return handler(true,"返回成功~");
        }
        // ================================消息队列==============================
        // 先将like行为进行处理，后续添加行为分析模块
        void HandleMessageForLike(std::unique_ptr<RdKafka::Message> msg){
            analytics::BehaviorActionEvent event;
            // 解析payload
            std::string payload(static_cast<const char*>(msg->payload()),msg->len());
            // 将payload反序列化为Protobuf对象
            if(event.ParseFromString(payload)){
                if(event.event_type() == analytics::LIKE){
                    // 更新点赞数
                    _user_mysql->update_userstatssummary_like(event.user_id());
                }else if(event.event_type() == analytics::UNLIKE){
                    // 取消点赞
                    _user_mysql->delete_userstatssummary_like(event.user_id());
                }
            }else{
                std::cerr << "[Kafka] Failed to parse protobuf message" << std::endl;
            }
        }
    };


    class UserServerBuild;
    class UserServer{
    public:
        UserServer(const std::shared_ptr<brpc::Server>& server,
                   const UserServiceImpl::ptr& user,
                   const Register::ptr& res)
            :_server(server),
            _user(user),
            _register(res)
        {}
        friend class UserServerBuild;
    public:
        using ptr = std::shared_ptr<UserServer>;
        void start (){
            _server->RunUntilAskedToQuit();
        }
    private:
        std::shared_ptr<brpc::Server> _server;
        UserServiceImpl::ptr _user;
        Register::ptr _register;
    };

    class UserServerBuild{
    public:
        UserServerBuild(){}
        
        UserServerBuild& make_odb_client(
            const std::string& user,
            const std::string& passwd,
            const std::string& db_name,
            const std::string& host,
            size_t port,
            const std::string& charset,
            size_t conn_pool_num
        ) {
            _odb = mysql_build::build(user, passwd, db_name, host, port, charset, conn_pool_num);
            return *this;
        }
        UserServerBuild& make_redis_client(const std::string& host,int port, int db_id,bool keepalive = true) {
            _redis = redis_build::build(host, port, db_id,keepalive);
            return *this;
        }
        // 启动消费线程
        UserServerBuild& make_consumer_thread(const std::string& brokers,const std::string& group_id,const std::string& topic) {
            _consumer = std::make_unique<KafkaConsumer>(brokers,group_id,topic);
            // 注册回调
            _consumer->setMessageCallback([&](std::unique_ptr<RdKafka::Message> msg){
                // 移交管理权
                _user->HandleMessageForLike(std::move(msg));
            });
            _consumer->start();
            return *this;
        }
        UserServerBuild& make_elasticsearch_client(const std::vector<std::string>& host){
            _esclient = std::make_shared<esclient>(host);
            return *this;
        }
        UserServerBuild& make_user_service() {
            if (!_odb) {
                errorlog << "failed to odb uninitizalier";
                abort();
            }
            if (!_redis) {
                errorlog << "failed to redis uninitizalier";
                abort();
            }
            if(!_esclient){
                errorlog << "failed to esclient uninitizalier";
                abort();
            }
            _user = std::make_shared<UserServiceImpl>(_odb,_redis,_esclient);
            return *this;
        }

        UserServerBuild& make_brpc_server(const std::size_t& port) {
            if(!_user){
                errorlog << "UserService uninitizalier";
                abort();
            }
            _server = std::make_shared<brpc::Server>();
            if(_server->AddService(&(*_user),brpc::SERVER_DOESNT_OWN_SERVICE) != 0){
                errorlog << "Failed to add service: UserService";
                abort();
            }
            brpc::ServerOptions options;
            options.idle_timeout_sec = -1;
            if(_server->Start(port,&options) != 0){
                errorlog << "Failed to start service: UserService";
                abort();
            }
            return *this;
        }
        UserServerBuild& make_register(const std::string& host,const std::string& value) {
            if(!_server){
                errorlog << "Can't register to etcd,because UserService not start";
                abort();
            }
            _register = std::make_shared<lyt::Register>(host);
            if(!_register->registory(lyt::user_path + value, value)){
                errorlog << "Fail register UserService to etcd";
                abort();
            }
            return *this;    
        }
        
        UserServer::ptr build() {
            if (!_server) {
                errorlog << "failed to brpc uninitizalier";
                abort();
            }
            if (!_user) {
                errorlog << "failed to user uninitizalier";
                abort();
            }
            if (!_register) {
                errorlog << "failed to register uninitizalier";
                abort();
            }
            return std::shared_ptr<UserServer>(new UserServer(_server,_user,_register));
        }

    private:
        shared_ptr<odb::mysql::database> _odb;
        std::shared_ptr<sw::redis::Redis> _redis ;
        std::shared_ptr<brpc::Server> _server;
        UserServiceImpl::ptr _user;
        lyt::Register::ptr _register;
        KafkaConsumer::ptr _consumer;
        esclient::ptr _esclient;
    };
}